package com.github.slprj.slake.net;

import com.github.slprj.slake.common.AbstractLifeCircle;
import com.github.slprj.slake.common.SlTime;
import com.github.slprj.slake.log.SlLog;
import com.github.slprj.slake.log.SlLogs;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 作为整个系统处理数据接收和发送的核心类,提供了以下几种功能
 * -建立SocketChannel连接,一个worker绑定了多个channel
 * -一個worker代表着一个线程
 * -主要处理读写请求的数据
 *
 * @author YangXiaoLong
 * 2022/8/23 11:25
 */
public class NioWorker extends AbstractLifeCircle implements Runnable {
    public static final SlLog log = SlLogs.getLogger(NioWorker.class);
    private static final int SS_RUNNING = 0;
    private static final int SS_PRE_STOP = 1;
    private static final int SS_STOP = 2;

    private final String workerName;
    private final Thread thread;
    private final NioStatus nioStatus;
    private final NioConfig config;
    private volatile int stopStatus = SS_RUNNING;
    private volatile Selector selector;
    private final AtomicBoolean notified = new AtomicBoolean(false);
    /**
     * ConcurrentLinkedQueue:线程安全的队列 , 支持先进先出,无锁操作
     * https://blog.csdn.net/qq_38293564/article/details/80798310
     * 构建新的serverSocket
     */

    private final ConcurrentLinkedQueue<Runnable> actions = new ConcurrentLinkedQueue<>();

    private int channelIndex;
    private final Collection<SlChannel> channels;
    private final IoQueue ioQueue;

    /**
     * 发送消息时,记录发送消息,收到结果时,删除消息
     */
    private final HashMap<Long, WriteData> pendingOutgoingRequests = new HashMap<>();
    private final CompletableFuture<Void> preCloseFuture = new CompletableFuture<>();

    private final Semaphore requestSemaphore;
    /**
     * 直接内存,每一个worker维护了一个属于自己的buffer
     */
    private final ByteBufferPool pool;

    private final WorkerParams workerParams;


    public NioWorker(NioStatus nioStatus, String workerName, NioConfig config) {
        this.nioStatus = nioStatus;
        this.config = config;
        this.thread = new Thread(this);
        this.workerName = workerName;
        this.thread.setName(workerName);
        this.requestSemaphore = nioStatus.getRequestSemaphore();

        if (config instanceof NioServerConfig) {
            this.channels = new HashSet<>();
            this.ioQueue = new IoQueue(null);
        } else {
            this.channels = new ArrayList<>();
            this.ioQueue = new IoQueue((ArrayList<SlChannel>) channels);
        }
        this.pool = new ByteBufferPool(true, config.getBufPoolSize(), config.getBufPoolMinCount(),
                config.getBufPoolMaxCount(), config.getBufPoolTimeout());
        RpcPbCallback pbCallback = new RpcPbCallback();
        workerParams = new WorkerParams();
        workerParams.setCallback(pbCallback);
        workerParams.setIoQueue(ioQueue);
        workerParams.setPendingRequests(pendingOutgoingRequests);
        workerParams.setWakeupRunnable(this::wakeup);
        workerParams.setPool(pool);

    }

    @Override
    protected void doStart() throws Exception {
        selector = SelectorProvider.provider().openSelector();
        thread.start();
    }

    public void preStop() {
        stopStatus = SS_PRE_STOP;
        wakeup();
    }

    @Override
    public void doStop() {
        stopStatus = SS_STOP;
        wakeup();
    }

    @Override
    public void run() {
        long cleanIntervalNanos = (long) config.getCleanIntervalMills() * 1000 * 1000;
        long lastCleanNanos = System.nanoTime();
        int selectTimeoutMillis = config.getSelectTimeoutMillis();
        while (this.stopStatus <= SS_PRE_STOP) {
            try {
                //本轮开始时间
                long roundStartTime = System.nanoTime();
                run0(selectTimeoutMillis, stopStatus, roundStartTime);
                //内核缓存区一直在忙,请求阻塞,消息一直发不出去
                if (roundStartTime - lastCleanNanos > cleanIntervalNanos) {
                    cleanTimeOutReq();
                    //byteBuffer池清理
                    lastCleanNanos = roundStartTime;
                }
            } catch (Exception e) {
                log.error("", e);
            }
        }
        //关闭
        try {
            selector.close();
            for (SlChannel slc : channels) {
                closeChannel0(slc.getChannel());
            }


            log.info("worker thread finished: {}", workerName);
        } catch (IOException e) {
            log.error("close error. {}", e);
        }
    }

    private void cleanTimeOutReq() {
        Iterator<Map.Entry<Long, WriteData>> it = this.pendingOutgoingRequests.entrySet().iterator();
        while (it.hasNext()) {
            Map.Entry<Long, WriteData> en = it.next();
            WriteData wd = en.getValue();

            SlTime t = wd.getTimeout();
            //请求已经过期
            if (t.rest(TimeUnit.MILLISECONDS) <= 0) {
                it.remove();
                log.debug("drop timeout request: {}ms,seq={},{}",
                        t.getTimeout(TimeUnit.MILLISECONDS),
                        wd.getData().seq,
                        wd.getSlc()
                );
                String msg = "timeout" + t.getTimeout(TimeUnit.MILLISECONDS) + "ms";
                wd.getFuture().completeExceptionally(new NetException(msg));
            }
            //释放Semaphore
            this.requestSemaphore.release();
        }
    }

    private void run0(int selectTimeoutMillis, int stopStatus, long roundStartTime) {
        if (!select(selectTimeoutMillis)) {
            return;
        }
        //执行newSocket的初始化
        performActions();
        //处理待写回客户端的消息
        boolean hasDataToWrite = ioQueue.dispatchWriteQueue(pendingOutgoingRequests);
        Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
        while (iterator.hasNext()) {
            hasDataToWrite |= processOneSelectionKey(iterator, stopStatus, roundStartTime);
        }
        //处于预关闭时,没有数据需要写回,则直接不需要等待了,返回结果
        if (stopStatus == SS_PRE_STOP) {
            if (!hasDataToWrite && pendingOutgoingRequests.size() == 0) {
                preCloseFuture.complete(null);
            }
        }
    }

    public void performActions() {
        //客户端:连接操作
        //服务端: 初始化socketChanel
        Runnable r;
        while ((r = actions.poll()) != null) {
            r.run();
        }
    }

    private boolean processOneSelectionKey(Iterator<SelectionKey> iterator, int stopStatus, long roundTime) {
        //https://blog.csdn.net/cold___play/article/details/106663776
        SelectionKey key = iterator.next();
        iterator.remove();
        SocketChannel sc = (SocketChannel) key.channel();
        int stage = 0;
        boolean hasDataToWrite = false;
        try {
            if (!key.isValid()) {
                log.info("socket may closed, remove it: {}", sc);
                //关闭通道
                closeChannel((SocketChannel) key.channel());
                return false;
            }
            //可连接状态
            stage = 1;
            if (key.isConnectable()) {
                processConnect(key);
                //处理
                return false;
            }

            stage = 2;
            SlChannel slc = (SlChannel) key.attachment();
            if (key.isReadable()) {
                ByteBuffer buf = slc.getOrCreateReadBuffer();
                //从sc中读取数据
                int readCount = sc.read(buf);
                if (readCount == -1) {
                    log.info("socket closed, remove it: {}", sc);
                    closeChannel((SocketChannel) key.channel());
                    return false;
                }
                slc.afterRead(stopStatus == SS_RUNNING);
            }

            stage = 3;
            //内核缓存区存在空闲时,则返回true;
            if (key.isWritable()) {
                IoSubQueue subQueue = slc.getSubQueue();
                //TODO 发送消息进行编码
                ByteBuffer buf = subQueue.getWriteBuffer(roundTime);
                if (buf != null) {
                    hasDataToWrite = true;
                    subQueue.setWriting(true);
                    //返回消息写入channel中
                    sc.write(buf);
                } else {
                    //没有数据可以写,则允许其他的channel写入数据
                    subQueue.setWriting(false);
                    key.interestOps(SelectionKey.OP_READ);
                }
            }
        } catch (Exception e) {
            //对不同阶段的异常进行日志输出
            switch (stage) {
                case 1:
                    log.warn("process socket connect error: {}", e.toString());
                    break;
                case 2:
                    log.warn("process socket read error: {}", e.toString());
                    break;
                case 3:
                    log.warn("process socket write error: {}", e.toString());
                    break;
                default:
                    log.warn("socket error: {}", e.toString());
            }
            closeChannel((SocketChannel) key.channel());
        }
        return hasDataToWrite;
    }

    /**
     * 处理就绪
     *
     * @param key 就绪事件
     * @author YangXiaoLong
     * 2022/10/17 17:09
     */
    @SuppressWarnings("all")
    private void processConnect(SelectionKey key) {
        Object[] att = (Object[]) key.attachment();
        //针对客户端而言
        CompletableFuture<SlChannel> f = (CompletableFuture<SlChannel>) att[0];
        Peer peer = (Peer) att[1];
        SocketChannel channel = (SocketChannel) key.channel();
        try {
            channel.finishConnect();
            SlChannel slc = initNewChannel(channel, peer);
            f.complete(slc);
        } catch (IOException e) {
            log.warn("init channel fail: {}, {}", channel, e.toString());
            closeChannel((SocketChannel) key.channel());
            f.completeExceptionally(new NetException(e));
        }
    }


    private boolean select(int selectTimeoutMillis) {
        try {
            //todo 等待过程中, 什么样的情况会导致失败
            if (selectTimeoutMillis > 0) {
                selector.select(selectTimeoutMillis);
            } else {
                selector.select();
            }
            return true;
        } catch (Exception e) {
            log.error("select failed: {}", workerName, e);
            return false;
        } finally {
            notified.set(false);
        }
    }

    private SlChannel initNewChannel(SocketChannel sc, Peer peer) throws IOException {
        sc.configureBlocking(false);
        //https://www.cnblogs.com/fnlingnzb-learner/p/16112827.html
        sc.setOption(StandardSocketOptions.SO_KEEPALIVE, false);
        //Nagle算法关闭后，客户端每发送一次数据，无论数据包的大小都会将这些数据发送出去。
        sc.setOption(StandardSocketOptions.TCP_NODELAY, true);

        SlChannel slc = new SlChannel(nioStatus, workerParams, config, sc, channelIndex++);
        SelectionKey selectionKey = sc.register(selector, SelectionKey.OP_READ, slc);
        slc.getSubQueue().setRegisterForWrite(new RegWriteRunner(selectionKey));
        if (peer != null) {
            peer.setSlChannel(slc);
            slc.setPeer(peer);
        }
        log.info("new DtChannel init: {}", sc);
        return slc;
    }

    public void newChannelAccept(SocketChannel sc) {
        actions.add(() -> {
            try {
                SlChannel slc = NioWorker.this.initNewChannel(sc, null);
                channels.add(slc);
            } catch (IOException e) {
                //失败
                log.warn("accept channel fail: {}, {}", sc, e.toString());
                NioWorker.this.closeChannel(sc);
            }
        });
        wakeup();
    }

    public CompletableFuture<SlChannel> connect(HostPort hp) {
        InetSocketAddress addr = new InetSocketAddress(hp.getHost(), hp.getPort());
        CompletableFuture<SlChannel> f = new CompletableFuture<>();
        actions.add(() -> {
            doConnect(addr, f, hp);
        });
        wakeup();
        return f;
    }

    private void doConnect(SocketAddress socketAddress, CompletableFuture<SlChannel> f, HostPort hp) {
        SocketChannel sc = null;
        try {
            sc = SocketChannel.open();
            sc.setOption(StandardSocketOptions.SO_KEEPALIVE, false);
            sc.setOption(StandardSocketOptions.TCP_NODELAY, true);
            sc.configureBlocking(false);
            sc.register(selector, SelectionKey.OP_CONNECT, new Object[]{f, hp});
            sc.connect(socketAddress);
        } catch (Throwable e) {
            if (sc != null) {
                closeChannel(sc);
            }
            f.completeExceptionally(new NetException(e));
        }
    }

    public void writeReqInBizThreads(Peer peer, WriteFrame frame, Decoder decoder,
                                     SlTime timeout, CompletableFuture<ReadFrame> future) {
        Objects.requireNonNull(timeout);
        Objects.requireNonNull(future);

        WriteData data = new WriteData(peer, frame, timeout, future, decoder);
        this.ioQueue.write(data);
        wakeup();
    }

    private void closeChannel(SocketChannel sc) {
        channels.removeIf(slc -> slc.getChannel() == sc);
        closeChannel0(sc);
    }

    private void closeChannel(SelectionKey key) {
        Object obj = key.attachment();
        SocketChannel sc = (SocketChannel) key.channel();
        if (obj instanceof SlChannel) {
            SlChannel slc = (SlChannel) obj;
            if (!slc.isClosed()) {
                slc.close();
                channels.remove(slc);
                closeChannel0(sc);
            }
        } else {
            channels.removeIf(d -> {
                if (d.getChannel() == sc) {
                    if (!d.isClosed()) {
                        d.close();
                        closeChannel0(sc);
                    }
                    return true;
                }
                return false;
            });
        }
    }

    private void closeChannel0(SocketChannel sc) {
        try {
            if (sc.isOpen()) {
                log.info("closing channel: {}", sc);
                sc.close();
            }
        } catch (IOException e) {
            log.warn("close channel fail: {}, {}", sc, e.getMessage());
        }
    }

    private void wakeup() {
        //nioWorker线程主动被唤醒
        if (Thread.currentThread() == thread) {
            return;
        }
        //防止多个线程并发做唤醒操作
        if (notified.compareAndSet(false, true)) {
            selector.wakeup();
        }
    }

    public Thread getThread() {
        return thread;
    }

    public String getWorkerName() {
        return workerName;
    }

    private static class RegWriteRunner implements Runnable {
        SelectionKey key;

        RegWriteRunner(SelectionKey key) {
            this.key = key;
        }

        @Override
        public void run() {
            key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
        }
    }
}
